package com.rrd.hive.udf.mexico;

import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.Text;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Field;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class SmsMakeFlagUDF extends UDF {

    public static Logger LOG = LoggerFactory.getLogger(SmsMakeFlagUDF.class);
    private final static String SPLIT = ",";

    public Text evaluate(String body) throws NoSuchAlgorithmException {
        UserSmsBody userSmsBody = new UserSmsBody();
        StringBuilder stringBuilder = new StringBuilder();
        Text text = new Text();
        final String finalBody =
                body.replaceAll("\r", " ")
                        .replaceAll("\n", " ")
                        .replaceAll(SPLIT, "")
                        .trim()
                        .toLowerCase();
        Arrays.stream(SmsFlag.values())
                .forEach(
                        flag -> {
                            final String pattern = flag.getPattern();
                            try {
                                if (pattern.contains(SPLIT)) {
                                    final String[] split = pattern.split(SPLIT);
                                    List<Integer> list =
                                            new ArrayList<>(split.length);
                                    for (String str : split) {
                                        list.add(hiveUdfRegExp(str, finalBody));
                                    }
                                    if (list.contains(0)) {
                                        setPoJoProperty(
                                                userSmsBody, flag.name(), 0);
                                    } else {
                                        setPoJoProperty(
                                                userSmsBody, flag.name(), 1);
                                    }
                                } else {
                                    setPoJoProperty(
                                            userSmsBody,
                                            flag.name(),
                                            hiveUdfRegExp(
                                                    flag.getPattern(), finalBody));
                                }
                            } catch (NoSuchFieldException
                                     | IllegalAccessException e) {
                                throw new RuntimeException(e);
                            }
                        });
        // 两个特殊的 flag
        userSmsBody.setFlag_submit(flag_submit(body));
        userSmsBody.setFlag_approve(flag_approve(body));
        // 基于上面 flag 的结果新增的 flag
        if (userSmsBody.getFlag_banknew() == 1
                && (userSmsBody.getFlag_pago() == 1
                || userSmsBody.getFlag_transfer() == 1
                || userSmsBody.getFlag_enviaste() == 1
                || userSmsBody.getFlag_recarga() == 1
                || userSmsBody.getFlag_ret() == 1
                || userSmsBody.getFlag_compraselect() == 1)) {
            userSmsBody.setFlag_expend(1);
        }

        if (userSmsBody.getFlag_banknew() == 1
                && (userSmsBody.getFlag_nomiadeposi() == 1
                || userSmsBody.getFlag_recib() == 1)) {
            userSmsBody.setFlag_income(1);
        }
        concatFlag(stringBuilder, userSmsBody);
        text.set(stringBuilder.toString());
        return text;
    }

    private int flag_submit(String body) {
        final int i = hiveUdfRegExp(SmsFlag.flag_loan.getPattern(), body);
        final int i1 = hiveUdfRegExp("exito|exitoso", body);
        final int i2 = hiveUdfRegExp("felicidades|solicitado", body);
        final int i3 = hiveUdfRegExp("solicitud recibida", body);
        if (i == 1 && ((i1 == 1 && i2 == 1) || i3 == 1)) {
            return 1;
        } else {
            return 0;
        }
    }
    private int flag_approve(String body) {
        final int i = hiveUdfRegExp(SmsFlag.flag_loan.getPattern(), body);
        final int i1 =
                hiveUdfRegExp(
                        "llego|sido aprobado|sido aprobada|se aprobo|fue aprobado|ha sido enviado|se ha enviado|aprobo la revisi|paso.*la revisi|la revisi.*paso|solicitado.*depositáis|depositáis.*solicitado|transferido.*cuenta|cuenta.*transferido|enviado.*cuenta|cuenta.*enviado|transfer|deposita|deposito",
                        body);
        final int i2 =
                hiveUdfRegExp(
                        "no ha sido aprobado|no has sido aprobado|no fue aprobado|fallida|se ha rechazado|no fue aprobado",
                        body);
        if (i == 1 && i1 == 1 && i2 == 0) {
            return 1;
        } else {
            return 0;
        }
    }

    private void concatFlag(StringBuilder stringBuilder, UserSmsBody userSmsBody) {
        final Class<UserSmsBody> userSmsBodyClass = UserSmsBody.class;
        final Field[] declaredFields = userSmsBodyClass.getDeclaredFields();
        for (Field declaredField : declaredFields) {
            declaredField.setAccessible(true);
            try {
                stringBuilder.append(declaredField.get(userSmsBody));
            } catch (IllegalAccessException e) {
                LOG.error("获取 UserSmsBody 对象的属性值报错了 {}", userSmsBody.toString());
            }
        }
    }
    public static int hiveUdfRegExp(String pattern, String input) {
        if (input == null) return 0;
        Pattern p = Pattern.compile(pattern);
        Matcher matcher = p.matcher(input);
        if (matcher.find(0)) {
            return 1;
        }
        return 0;
    }

    private <T> void setPoJoProperty(T t, String name, T value)
            throws NoSuchFieldException, IllegalAccessException {
        final Class<?> aClass = t.getClass();
        final Field declaredField = aClass.getDeclaredField(name);
        declaredField.setAccessible(true);
        declaredField.set(t, value);
    }
}
